﻿#include "cross_com.h"

#include "dvr_mml_svr.h"
#include "dvr_cmd.h"
#include "MultiCHData.h"

#ifdef _WIN32

#include <winsock2.h>
#else
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <sys/ioctl.h>
#include "media_frame_hd.h"
#include "storage_api.h"
#endif

//支持4个用户

int CStreamerSvr::LIVE_SEND_BIT = 0xF;

CStreamerSvr::CStreamerSvr()

{
	m_rcvBuf = NULL;
	m_sendBuf = NULL;
	if(NULL == m_rcvBuf)
	{
		m_rcvBuf_sz = SIZE_KB(8);
		m_rcvBuf = new unsigned char[m_rcvBuf_sz];
	}

	m_send_sz = NULL;

	m_sock    = 0;
	memset(m_rcvBuf,0,SIZE_KB(4));

	m_live_ch = 0;
	m_bStream_Live = false;
	m_bReplay = false;
	m_bSendSvr = true;
	m_bMix_ch = false;
#ifdef _WIN32
	WORD sockVersion = MAKEWORD(2,2);
	WSADATA wsaData;
	static bool bInit = false;

	if(!bInit)
	{
		if(WSAStartup(sockVersion, &wsaData)!=0)
		{
			return ;
		}
		bInit = true;
	}

#endif
}

CStreamerSvr::~CStreamerSvr()
{

}
int CStreamerSvr::Init(int send_buf_sz,int portNum,bool bSend)
{
	if(NULL == m_rcvBuf)
	{
		m_rcvBuf = new unsigned char[SIZE_KB(4)];
	}

	if(NULL == m_sendBuf)
	{
		m_send_sz = send_buf_sz;
		m_sendBuf  = new char[send_buf_sz];
	}
	memset(m_rcvBuf,0,SIZE_KB(4));
	memset(m_sendBuf,0,m_send_sz);

	//创建套接字
	m_sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	if(m_sock == INVALID_SOCKET)
	{
		 printf("socket error !");
		 return 0;
	}
	int try_cnt = 0;
	int on = 1;
	while (true)
	{
		int ret = setsockopt(m_sock,SOL_SOCKET,SO_REUSEADDR,(const char*)&on,sizeof(on));
		if(ret < 0)
		{
			 printf("set sockopt fail.!");
		}
		//绑定IP和端口
		sockaddr_in sin;
		sin.sin_family = AF_INET;
		sin.sin_port = htons(portNum);
	#ifdef _WIN32
		sin.sin_addr.S_un.S_addr = INADDR_ANY;
	#else
		sin.sin_addr.s_addr = htonl(INADDR_ANY);
	#endif // _WIN32
		if(bind(m_sock, (struct sockaddr *)&sin, sizeof(sin)) == -1)
		{
			TRACE("bind error %d! m_sock %d try_cnt %d \n",errno,m_sock,try_cnt);
			if(try_cnt >= 3)
			{
				return -1;
			}
			try_cnt++;
			CrossSleep(1000);
			continue;
		}
		else
		{
			break;
		}
	};

	int i = 0;
	for(i = 0; i < 4; i++)
	{
		m_stream_client[i].clientID   = 0;
		m_stream_client[i].clientSock = -1;
		m_stream_client[i].pTask      = this;
		m_stream_client[i].stream_ch  = 0;
		m_stream_client[i].frmBuf  = new unsigned char[500 << 10];
	}
	m_bSendSvr = bSend;
    TRACE("Init Stream svr Ok m_sock %d .\n",m_sock);
	return 0;
}

#ifdef _WIN32

int CStreamerSvr::ListenProc()
{
	int max_sock = 0;
	int strLen;
	fd_set  m_fds,copyRead;
	struct timeval tm;
	unsigned long ul=1;
	SOCKADDR_IN clntAddr;
	int clntAddrSz;
	int read_byte = 0;
	int yes = 1;

	m_sockNum = 0;

	max_sock = m_sock;

	if(listen(m_sock,SOCK_SIZE)==-1)
	{
		TRACE_ERR("listen error,err %d ",errno);
		return -1;
	}
	//遍历socks，将所有的元素置于无效的socket
	for(int i=0;i<SOCK_SIZE;i++)
		m_send_sock[i] = INVALID_SOCKET;
	m_sockNum += 1;

	m_send_sock[0]=m_sock;
	FD_ZERO(&m_fds);
	FD_SET(m_sock,&m_fds);
	tm.tv_sec=5000;
	tm.tv_usec=0;
    TRACE_ERR("Strat  MML server accpet client \n ");

	while(1)
	{
		copyRead=m_fds;
		int selResult = select(m_sockNum +1,&copyRead,0,0,&tm);
		if(selResult==-1)
		{
			puts("select error");
		}
		else if(selResult==0)
		{
			//TRACE_ERR("Time out .\n");
			continue;
		}
		else
		{
			//先判断是否是有新的客户端连接
			if(FD_ISSET(m_send_sock[0],&copyRead))
			{
				clntAddrSz = sizeof(clntAddr);
				int clntSock = accept(m_sock,(struct sockaddr*)&clntAddr,&clntAddrSz);
				//将socket设置成非阻塞模式
				ioctlsocket(clntSock,FIONBIO,&ul);

				m_lock.Lock();
				for(int i = 0;i<SOCK_SIZE;i++)
				{
					//遍历socks，在元素为无效的socket处插入客户端的socket
					if(m_send_sock[i] == INVALID_SOCKET)
					{
						FD_SET(clntSock,&m_fds);
						TRACE("i %d accept new sokcet %d .\n",i,clntSock);
						int ret = StartStreamSvr(clntSock,i);
						if(ret == ERR_OVER_MAXCLIENT)
						{   //超最大用户支持数
#ifdef _WIN32
							closesocket(clntSock);
#else
							close(clntSock);
#endif	
						}
						else
						{
							m_send_sock[i] = clntSock;
							m_sockNum++;
						}
						break;
					}
				}
				m_lock.UnLock();
			}
			//遍历所有的客户端socket，0的位置为服务端的socket，所以从1开始
			for (int i = 1;i < SOCK_SIZE;i++)
			{
				//如果是无效的socket 不必处理
				if(m_send_sock[i] == INVALID_SOCKET)
					continue;
				if(FD_ISSET(m_send_sock[i],&copyRead))
				{
					strLen=recv(m_send_sock[i],(char *)m_rcvBuf,m_rcvBuf_sz,0);
					if(strLen <= 0)//客户端断开了连接
					{
						closesocket(m_send_sock[i]);
						//从fds删除客户端socket
						FD_CLR(m_send_sock[i],&m_fds);
						//将对应的位置再次置为无效socket
						m_send_sock[i] = INVALID_SOCKET;
						m_sockNum--;
					}
				}
			}
		}
	}
	closesocket(m_sock);
	return 0;
}
#else
int CStreamerSvr::ListenProc()
{
	int max_sock = 0;
	fd_set copyRead;
	struct timeval tm;
	unsigned long ul=1;
	struct sockaddr_in clntAddr;
	socklen_t clntAddrSz;
	int i = 0;

	m_sockNum = 0;

	max_sock = m_sock;

	if(listen(m_sock,SOCK_SIZE)==-1)
	{
		TRACE_ERR("listen error,m_sock %d err %d ",m_sock,errno);
		return -1;
	}
	//遍历socks，将所有的元素置于无效的socket
	for(int i=0;i<SOCK_SIZE;i++)
		m_send_sock[i] = INVALID_SOCKET;

	m_sockNum = m_sock;
	m_send_sock[0]=m_sock;

	tm.tv_sec=5000;
	tm.tv_usec=0;
	TRACE_ERR("Strat accpet client \n ");
	while(1)
	{
		FD_ZERO(&m_fds);
		FD_SET(m_sock,&m_fds);

		max_sock = m_sock;
		//遍历socket池 找出值最大的socket
		for (i = 0; i < SOCK_SIZE; i++)
		{
			if (m_send_sock[i] != -1)
			{
				//将socket池中的所有socket都添加到事件数组allset中
				FD_SET(m_send_sock[i], &m_fds);
				if (m_send_sock[i] > max_sock)
				{
					max_sock = m_send_sock[i];    //maxfd永远是值最大的socket
				}
			}
		}

		int selResult = select(m_sock +1,&m_fds,0,0,&tm);
		if(selResult < 0)
		{
			printf("select failed ! max_sock %d  error message:%s\n", max_sock,strerror(errno));
			break;
		}
		else if(selResult==0)
		{
		    continue;
		}
		else
		{
			//先判断是否是有新的客户端连接
			if(FD_ISSET(m_send_sock[0],&m_fds))
			{
				clntAddrSz = sizeof(clntAddr);
				int clntSock = accept(m_sock,(struct sockaddr*)&clntAddr,&clntAddrSz);
				//将socket设置成非阻塞模式

				ioctl(clntSock,FIONBIO,&ul);
				for(int i=0;i<SOCK_SIZE;i++)
				{
					//遍历socks，在元素为无效的socket处插入客户端的socket
					if(m_send_sock[i] == INVALID_SOCKET)
					{
						FD_SET(clntSock,&m_fds);
						m_send_sock[i]=clntSock;
						m_sockNum++;
						StartStreamSvr(clntSock);
						break;
					}
				}
			}

			//遍历所有的客户端socket，0的位置为服务端的socket，所以从1开始
			for (int i=1;i < SOCK_SIZE;i++)
			{
				//如果是无效的socket 不必处理
				if(m_send_sock[i]==INVALID_SOCKET)
					continue;
//				if(FD_ISSET(m_send_sock[i],&copyRead))
//				{
//					strLen=recv(m_send_sock[i],(char *)m_rcvBuf,m_rcvBuf_sz,0);
//					if(strLen <= 0)//客户端断开了连接
//					{
//
//						close(m_send_sock[i]);
//						//从fds删除客户端socket
//						FD_CLR(m_send_sock[i],&fds);
//						//将对应的位置再次置为无效socket
//						m_send_sock[i] == INVALID_SOCKET;
//						m_sockNum--;
//					}
//				}
			}
		}
	}

	close(m_sock);
	return 0;
}
#endif
int CStreamerSvr::StartStreamSvr(SOCKET clntSock,int index)
{
	int ret = 0;

	STREAM_SVR_INFO_S * pSvrInfo = GetFreeClientInfo(clntSock,index);
	if(pSvrInfo == NULL)
	{   //超最大用户支持数
		TRACE_ERR("No Free clientID.\n");
		return ERR_OVER_MAXCLIENT;
	}
	if(m_bSendSvr)
	{
		m_dataThdID = Cross_CreateThread((cross_start_routine) StreamProc, (void *)pSvrInfo,NULL, SIZE_128KB);
	}
	else
	{
		m_dataThdID = Cross_CreateThread((cross_start_routine) StreamProcRecv, (void *)pSvrInfo,NULL, SIZE_128KB);
	}
	

	return ret;
}


int CStreamerSvr::RcvData(char * buf ,int buf_sz,int conn_fd)
{
	fd_set      mFdr;
	timeval  mTimeOut;

	FD_ZERO( &mFdr );
	FD_SET( conn_fd, &mFdr );
	int ret = 0;
	mTimeOut.tv_sec = 1;
	mTimeOut.tv_usec = 100000;
	int flags = select( conn_fd + 1, &mFdr, NULL, NULL, &mTimeOut );
	if( flags > 0 )
	{
		ret = recv(conn_fd, buf, buf_sz, 0);
		if(ret == 0 )
		{
			TRACE_ERR("Can not recive data. clientSock %d \n",conn_fd);
			return 0;
		}

	}
	return ret;
}

int CStreamerSvr::SendData(unsigned char* buf, int len, int flags,int conn_fd)
{
	int sendBytes = 0;
	fd_set  mFdw;
	FD_ZERO( &mFdw );
	FD_SET( conn_fd, &mFdw );
	timeval mTimeOut;
	mTimeOut.tv_sec = 0;
	mTimeOut.tv_usec = 5*1000;
	select( conn_fd + 1, NULL, &mFdw, NULL, &mTimeOut );
	if ( !FD_ISSET(conn_fd, &mFdw) )
	{
		return SOCKETNODATA;
	}

	sendBytes = send(conn_fd, (char*)buf, len, flags);

	return sendBytes;
}

int  CStreamerSvr::ReadFrameData(unsigned char * pFrm,int buf_sz,int ch, int clientID)
{
	int ret = 0;
	if(m_bReplay)
	{
		ret = REPLAY_FIFO_READ_FRAME((unsigned char *)pFrm,buf_sz,ch,clientID);
	}
	else
	{
#ifdef _WIN32
		CMultiCHData* pMulData = CMultiCHData::getSingleton();

		ret = pMulData->FIFO_GET_FRAME((unsigned char *)pFrm,buf_sz,ch,clientID); // MML_Read_Frame(m_live_ch,m_pBuf,buf_size,CLIENT_ID_LIVE)
#else
		ret = MML_Read_Frame((unsigned char *)pFrm,buf_sz,ch,2);
#endif
	}
	return ret;
}

STREAM_SVR_INFO_S * CStreamerSvr::GetFreeClientInfo(int sock,int index)
{
	int i = 0;

	for(i = 0; i < 4; i++)
	{
		if(-1 == m_stream_client[i].clientSock)
		{
			m_stream_client[i].clientSock = sock;
			m_stream_client[i].clientID  = 2; //起始ID设计为2，1为RTSP
			m_stream_client[i].pTask     = (void*)this;
			m_stream_client[i].stream_ch = 0;
			m_stream_client[i].sock_idx = index;
			m_stream_client[i].bMix_ch = false;
			m_stream_client[i].send_ch_bit = 0xF; //默认4通道
			m_stream_client[i].last_ch = 0;
			return &m_stream_client[i];
		}
	}

	return NULL;
}

void CStreamerSvr::DestroyClient(int sock)
{
	int i = 0;

	for(i = 0; i < 4; i++)
	{
		if(sock == m_stream_client[i].clientSock)
		{
			m_stream_client[i].clientSock = -1;
			break;
		}
	}

}

int GetNexCh(int & startCh,int chBit)
{  
	int dest_ch = 0;
	int i = 0;
	bool bFound  = false;
	if(startCh == 31)
	{
		startCh = 0;
	}

	for(i = startCh + 1 ; i < 16 ; i++)
	{
		if((chBit>>i) & 0x1)
		{
			dest_ch = i;
			bFound = true;
			break;
		}
	}

	if(!bFound)
	{
		for(i = 0; i < 16 ; i++)
		{
			if((chBit>>i) & 0x1)
			{
				dest_ch = i;
				bFound = true;
				break;
			}
		}
	}

	startCh = dest_ch;
    return dest_ch;
}

int CStreamerSvr:: StreamProc(void * pParam)
{
	STREAM_SVR_INFO_S* pClientSock = (STREAM_SVR_INFO_S * )pParam;
	CStreamerSvr * pTask = (CStreamerSvr*)pClientSock->pTask;
#ifdef _WIN32
	SOCKET conn_fd;
#else
	int conn_fd;
#endif

	conn_fd = pClientSock->clientSock;
	unsigned char * pFrm = pClientSock->frmBuf;
	int    ret  = 0;
	int    frmLen = 0;
	int    buf_sz = pTask->m_send_sz;
	int    data_pos  =0;
	int    err = 0;
	struct timeval tmpT;
	int64_t curTm;
	LHN_FRAME_H * pHead = (LHN_FRAME_H*)pFrm;

	pTask->m_bStream_Live = true;
	int dest_ch = 0;
	TRACE("Start stream proc thread.\n");

	while (pTask->m_bStream_Live)
	{

		dest_ch = GetNexCh(pClientSock->last_ch,CStreamerSvr::LIVE_SEND_BIT);

		frmLen = ret = pTask->ReadFrameData(pFrm,buf_sz,dest_ch,pClientSock->clientID);

		if(ret < 0)
		{
			CrossSleep(30);
			continue;
		}
		gettimeofday(&tmpT, NULL);//时间戳
		curTm = (int64_t)tmpT.tv_sec * 1000 + (int64_t)tmpT.tv_usec/1000;
		int64_t diff = curTm - pHead->pts;
		if( pHead->frameType == IFrame && diff > 90 )
		{
			TRACE("ch %02d FNO %d Send Frame rate %d len %d diff %lld curTm %lld pts %lld  \n",pHead->channel,pHead->frameNo,
				pHead->frameRate,pHead->dataLen,diff,curTm,pHead->pts);
		}

		if(frmLen > buf_sz)
		{
			continue;
		}

		data_pos = 0;
		int send_len = 0;
		int i = 0;
		int leaf_len = 0;

		while (data_pos < frmLen)
		{
			int try_cnt = 3;
			bool bSendOk = false;

            leaf_len = frmLen - data_pos;

			for(i = 0; i < try_cnt; i++)
			{
				ret = pTask->SendData(pFrm + data_pos,leaf_len,0,conn_fd);

				if(ret >= 0)
				{
					bSendOk = true;
					break;
				}
				if(ret <= 0)
				{
					err = errno;
					if(err == 32)
					{
						goto SEND_DONE;
					}
					TRACE("Send frame  last error %d conn_fd %d ret %d send_len %d  frmLen %d leaf_len %d .\n",
						   err,conn_fd,ret,send_len,frmLen,leaf_len);
					if(i == try_cnt - 1)
					{
						goto SEND_DONE;
					}
				}
			}
			if(!bSendOk)
			{
				break;
			}
			send_len += ret;
			if(send_len  >= frmLen)
			{
				break;
			}
			data_pos += ret;
		}

        CrossSleep(1);
	}

SEND_DONE:
#ifdef _WIN32
	closesocket(conn_fd);
#else
	close(conn_fd);
	int idx = pTask->m_sockNum;
	//从fds删除客户端socket
//	FD_CLR(pTask->m_send_sock[idx],&pTask->m_fds);
	//将对应的位置再次置为无效socket
	pTask->m_send_sock[idx -1 ] = INVALID_SOCKET;
	pTask->m_sockNum--;
#endif
	pTask->DestroyClient(conn_fd);
	TRACE("Exit stream thread.\n");

	return 0;
}


int CStreamerSvr::StreamProcRecv(void * pParam)
{
	STREAM_SVR_INFO_S* pClientSock = (STREAM_SVR_INFO_S * )pParam;
	CStreamerSvr * pTask = (CStreamerSvr*)pClientSock->pTask;
#ifdef _WIN32
	SOCKET conn_fd;
#else
	int conn_fd;
#endif

	conn_fd = pClientSock->clientSock;
	unsigned char * pFrm = pClientSock->frmBuf;
	int    ret  = 0;
	int    pktLen = 0;
	int    buf_sz = pTask->m_send_sz;
	int    data_pos  =0;
	int    err = 0;
	int    try_cnt = 3;

	LHN_FRAME_H * pHead = (LHN_FRAME_H*)pFrm;

	pTask->m_bStream_Live = true;

	TRACE("Start stream proc thread.\n");
	CMultiCHData* mulData = CMultiCHData::getSingleton();
	while (pTask->m_bStream_Live)
	{
		
		pktLen = ret = pTask->RcvData((char*)pFrm,buf_sz,conn_fd);

		if(ret <0 )
		{
			if( 0 == try_cnt--)
			{
				goto SEND_DONE;
			}
			TRACE("Rcv ret %d err %d .\n",ret,errno);
			CrossSleep(10);
		}
		if(ret > 0)
		{
			ret = mulData->FIFO_WRITE_FRAME(pFrm,pktLen,0,0,0,true);
			TRACE("Rcv data size %d .\n",pktLen);
		}

		CrossSleep(30);
	}

SEND_DONE:
	pTask->m_lock.Lock();
	int idx = pClientSock->sock_idx;
	if(pTask->m_send_sock[idx] !=INVALID_SOCKET )
	{
#ifdef _WIN32
		closesocket(conn_fd);
#else
		close(conn_fd);
#endif

		//从fds删除客户端socket
		//	FD_CLR(pTask->m_send_sock[idx],&pTask->m_fds);
		//将对应的位置再次置为无效socket
		pTask->m_send_sock[idx] = INVALID_SOCKET;
		pTask->m_sockNum--;

		pTask->DestroyClient(conn_fd);
		
	
	}
	pTask->m_lock.UnLock();
	TRACE("Exit stream thread.\n");
	return 0;
}
